package com.lry.core;

import com.lry.entity.AppendEntriesRequest;
import com.lry.entity.AppendEntriesResponse;
import com.lry.entity.VoteRequest;
import com.lry.entity.VoteResponse;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicReference;

//https://hardcore.feishu.cn/docs/doccnMRVFcMWn1zsEYBrbsDf8De
//https://github.com/maemual/raft-zh_cn/blob/master/raft-zh_cn.md
@Slf4j
@Getter
@Setter
public class RaftNode {


    //如果commitIndex > lastApplied，则 lastApplied 递增，并将log[lastApplied]应用到状态机中（5.3 节）
    //todo 日志复制这块问题还很多，commitIndex,lastApplied,nextIndex,matchIndex这四个变量的维护还很多问题
    //假设存在 N 满足N > commitIndex，使得大多数的 matchIndex[i] ≥ N以及log[N].term == currentTerm 成立，则令 commitIndex = N

    public enum NodeState {
        STATE_FOLLOWER,
        STATE_PRE_CANDIDATE,
        STATE_CANDIDATE,
        STATE_LEADER
    }

    private StateMachine stateMachine;

    private List<Server> servers;

    private List<Peer> peers;
    private Map<Integer,Peer> serverId2Peer;

    private Server localServer;
    private Integer serverId;

    private Integer leaderId;

    private volatile NodeState state = NodeState.STATE_FOLLOWER;

    //服务器当前的任期，逻辑时钟
    private long currentTerm;

    //投票给了谁
    private volatile Integer votedFor;

    //log[]
    private List<Log> logs;

    //如果commitIndex > lastApplied，则 lastApplied 递增，并将log[lastApplied]应用到状态机中
    //已知已提交的最高的日志条目的索引（初始值为0，单调递增）
    private long commitIndex;

    //已经被应用到状态机的最高的日志条目的索引（初始值为0，单调递增）
    private long lastApplied;

    //用于节点间通信的线程池
    private ExecutorService executorService;

    //定时任务
    private ScheduledExecutorService scheduledExecutorService;

    //选举定时器
    private ScheduledFuture electionScheduledFuture;

    //心跳定时器
    private ScheduledFuture heartbeatScheduledFuture;

    //raft配置
    private RaftOptions raftOptions;


    public RaftNode(RaftOptions raftOptions,Server localServer,List<Server> servers,StateMachine stateMachine) {
        assert null!=localServer;
        assert null!=servers && !servers.isEmpty();
        assert null!=raftOptions;
        assert null!=stateMachine;
        this.raftOptions = raftOptions;
        this.stateMachine = stateMachine;
        this.localServer = localServer;
        this.serverId = localServer.getServerId();
        this.servers = servers;
    }

    public void start() {
        logs = new ArrayList<>();
        peers = new ArrayList<>();
        serverId2Peer = new ConcurrentHashMap<>();

        for (Server server : this.servers) {
            if(server.equals(localServer)){
                continue;
            }
            Peer peer = new Peer(server);
            serverId2Peer.put(server.getServerId(),peer);
            peers.add(peer);
        }

        // init thread pool
        executorService = new ThreadPoolExecutor(
                raftOptions.getRaftConsensusThreadNum(),
                raftOptions.getRaftConsensusThreadNum(),
                60,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<Runnable>());

        scheduledExecutorService = Executors.newScheduledThreadPool(2);

        //快照
        scheduledExecutorService.scheduleWithFixedDelay(new Runnable() {
            @Override
            public void run() {
                takeSnapshot();
            }
        }, raftOptions.getSnapshotPeriodSeconds(), raftOptions.getSnapshotPeriodSeconds(), TimeUnit.SECONDS);

        // start election
        resetElectionTimer();
    }


    private void takeSnapshot() {

    }


    private void resetHeartbeatTimer() {
        if (heartbeatScheduledFuture != null && !heartbeatScheduledFuture.isDone()) {
            heartbeatScheduledFuture.cancel(true);
        }
        heartbeatScheduledFuture = scheduledExecutorService.schedule(new Runnable() {
            @Override
            public void run() {
                startNewHeartbeat();
            }
        }, raftOptions.getHeartbeatPeriodMilliseconds(), TimeUnit.MILLISECONDS);
    }

    private void startNewHeartbeat() {
        log.info("server={},发送心跳给所有peer",serverId);
        for (Peer peer : peers) {
            appendEntries(peer);
        }
        resetHeartbeatTimer();
    }

    public long getLastLogIndex(){
        if(logs.size()<1){
            return 0;
        }
        return logs.get(logs.size()-1).getIndex();
    }


    private long getEntryTerm(long index){
        for (Log log1 : logs) {
            if(index==log1.getIndex()){
                return log1.getTerm();
            }
        }
        return 0;
    }

    public byte[] get(byte[] command) {
        //追加到日志条目中
        Log log = new Log(command,currentTerm,getLastLogIndex()+1);
        Log[] logs = new Log[1];
        logs[0] = log;

        List<Future<AppendEntriesResponse>> futures = new ArrayList<>();
        //同步给其他节点
        for (Peer peer : peers) {

            long prevLogIndex = peer.getNextIndex() -1;
            long prevLogTerm = getEntryTerm(prevLogIndex);

            AppendEntriesRequest appendEntriesRequest = AppendEntriesRequest.builder().
                    term(currentTerm).
                    leaderId(serverId).
                    preLogTerm(prevLogTerm).
                    preLogIndex(prevLogIndex).
                    entries(logs).
                    leaderCommit(commitIndex).build();

            Future<AppendEntriesResponse> future = peer.getRaftConsensusServiceAsync().get(appendEntriesRequest, new RpcCallback<AppendEntriesResponse>() {
                @Override
                public void success(AppendEntriesResponse obj) {

                }

                @Override
                public void fail(Throwable throwable) {

                }
            });
            futures.add(future);
        }

        int replicateSuccessNum = 0;
        for (Future<AppendEntriesResponse> future : futures) {
            try {
                AppendEntriesResponse appendEntriesResponse = future.get();
                if(appendEntriesResponse.isSuccess()){
                    replicateSuccessNum++;
                    if(replicateSuccessNum+1>servers.size()/2){
                        return stateMachine.get(command);
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
        return null;
    }

    public int errorNum = 0;

    public byte[] replicate(byte[] command) {
        errorNum++;
        //追加到日志条目中
        Log log = new Log(command,currentTerm,getLastLogIndex()+1);
        logs.add(log);

//        if(errorNum<=1){
//            return true;
//        }

        List<Future<AppendEntriesResponse>> futures = new ArrayList<>();
        //同步给其他节点
        for (Peer peer : peers) {

            long prevLogIndex = peer.getNextIndex() -1;
            long prevLogTerm = getEntryTerm(prevLogIndex);


            AppendEntriesRequest appendEntriesRequest = AppendEntriesRequest.builder().
                    term(currentTerm).
                    leaderId(serverId).
                    preLogTerm(prevLogTerm).
                    preLogIndex(prevLogIndex).
//                    entries(new Log[]{log}).
                    entries(nextIndexLogs(peer.getNextIndex())).
                    leaderCommit(commitIndex).
                    build();

            Future<AppendEntriesResponse> future = peer.getRaftConsensusServiceAsync().appendEntries(appendEntriesRequest,
                    new AppendEntriesResponseCallBack(appendEntriesRequest,peer));

            futures.add(future);
        }

        int replicateSuccessNum = 0;
        for (Future<AppendEntriesResponse> future : futures) {
            try {
                AppendEntriesResponse appendEntriesResponse = future.get();
                if(appendEntriesResponse.isSuccess()){
                    replicateSuccessNum++;
                    if(replicateSuccessNum+1>servers.size()/2){
                        commitIndex++;
                        lastApplied++;
                        return  stateMachine.apply(command);
                    }
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }

        return null;
    }

    private void appendEntries(Peer peer) {
        AppendEntriesRequest appendEntriesRequest = AppendEntriesRequest.builder().
                term(currentTerm).
                leaderId(serverId).
                preLogTerm(0).
                preLogIndex(0).
                build();

         peer.getRaftConsensusServiceAsync().appendEntries(appendEntriesRequest,
                new AppendEntriesResponseCallBack(appendEntriesRequest,peer));

    }

    public void resetElectionTimer() {
        //选举定时器没走完就取消掉，因为收到心跳包需要调用这里，发现没走完就要取消定时器
        if (electionScheduledFuture != null && !electionScheduledFuture.isDone()) {
            log.info("Cancel election timer");
            electionScheduledFuture.cancel(true);
        }
        electionScheduledFuture = scheduledExecutorService.schedule(new Runnable() {
            @Override
            public void run() {
                startVote();
            }
        }, getElectionTimeoutMs(), TimeUnit.MILLISECONDS);

    }

    private int getElectionTimeoutMs() {
        ThreadLocalRandom random = ThreadLocalRandom.current();
        int randomElectionTimeout = raftOptions.getElectionTimeoutMilliseconds() + random.nextInt(0, raftOptions.getElectionTimeoutMilliseconds());
        log.debug("server={} ,new election time is after {} ms", serverId,randomElectionTimeout);
        return randomElectionTimeout;
    }


    public Log getLastLog(){
        if(logs.size()<1){
            return null;
        }
       return logs.get(logs.size()-1);
    }

    private void startVote() {
        log.info("server={},Running start-vote in term {}",serverId, currentTerm);

        currentTerm++;
        state = NodeState.STATE_CANDIDATE;
        votedFor = serverId;
        leaderId = null;

        Log lastLog = getLastLog();

        for (Peer peer : peers) {

            peer.setVoteGranted(null);
            VoteRequest voteRequest = VoteRequest.builder().
                    term(currentTerm).
                    candidateId(serverId).
                    lastLogTerm(lastLog!=null?lastLog.getTerm():0).
                    lastLogIndex(lastLog!=null?lastLog.getIndex():0).
                    build();

            peer.getRaftConsensusServiceAsync().requestVote(voteRequest,new VoteResponseCallBack(peer,voteRequest));
        }

//        resetElectionTimer();
    }


    private class AppendEntriesResponseCallBack implements RpcCallback<AppendEntriesResponse> {

        private Peer peer;
        private AppendEntriesRequest request;

        public AppendEntriesResponseCallBack(AppendEntriesRequest request,Peer peer){
            this.peer = peer;
            this.request = request;
        }

        @Override
        public void success(AppendEntriesResponse appendEntriesResponse) {
            //如果收到的响应的任期比我大，直接成为follower
            if (appendEntriesResponse.getTerm() > currentTerm) {

                log.info("server={},Received response from server {} " +
                                "in term {} (this server's term was {})", serverId,
                        peer.getServer().getServerId(),
                        appendEntriesResponse.getTerm(),
                        currentTerm);

                stepDown(appendEntriesResponse.getTerm());
            }else{
                //失败了，减少nextIndex 重试，直到日志匹配
                if(!appendEntriesResponse.isSuccess()){
                    peer.decrNextIndex();

                    long prevLogIndex = peer.getNextIndex() - 1;
                    long prevLogTerm = getEntryTerm(prevLogIndex);


                    AppendEntriesRequest appendEntriesRequest = AppendEntriesRequest.builder().
                            term(currentTerm).
                            leaderId(serverId).
                            preLogTerm(prevLogTerm).
                            preLogIndex(prevLogIndex).
                            entries(nextIndexLogs(peer.getNextIndex())).
                            leaderCommit(commitIndex).
                            build();

                    Future<AppendEntriesResponse> future = peer.getRaftConsensusServiceAsync().appendEntries(appendEntriesRequest,
                            new AppendEntriesResponseCallBack(appendEntriesRequest,peer));
                }else{
                    //更新跟随者的nextIndex
                    peer.setNextIndex(request.getPreLogIndex()+1+(request.getEntries()!=null?request.getEntries().length:0));
                    peer.setNextIndex(request.getPreLogIndex()+(request.getEntries()!=null?request.getEntries().length:0));
                }
            }
        }

        @Override
        public void fail(Throwable throwable) {

        }
    }

    private Log[] nextIndexLogs(long nextIndex){
        if(logs.isEmpty()){
            return null;
        }
        List<Log> list = new ArrayList<>();

        for(int i=logs.size()-1;i>=0;i--){
            Log log1 = logs.get(i);
            list.add(0,log1);
            if(log1.getIndex()==nextIndex){
                break;
            }
        }
        return list.toArray(new Log[list.size()]);
    }

    private class VoteResponseCallBack implements RpcCallback<VoteResponse> {

        private VoteRequest voteRequest;
        private Peer peer;

        public VoteResponseCallBack(Peer peer,VoteRequest voteRequest) {
            this.peer = peer;
            this.voteRequest = voteRequest;
        }

        @Override
        public void success(VoteResponse voteResponse) {

            peer.setVoteGranted(voteResponse.isVoteGranted());

            //如果当前任期和请求时候的任期不一致 || 状态不是候选人 放弃成为leader的可能性
            if (currentTerm != voteRequest.getTerm() || state != NodeState.STATE_CANDIDATE) {
                log.info("ignore requestVote RPC result");
                return;
            }
            //如果收到的响应的任期比我大，我要放弃竞选，直接成为follower
            if (voteResponse.getTerm() > currentTerm) {
                log.info("Received RequestVote response from server {} " +
                                "in term {} (this server's term was {})",
                        peer.getServer().getServerId(),
                        voteResponse.getTerm(),
                        currentTerm);
                stepDown(voteResponse.getTerm());
            } else {
                //如果返回成功，则统计票数，过半成为leader
                if(voteResponse.isVoteGranted()){

                    log.info("server {} Got vote from server {} for term {}",serverId,
                            peer.getServer().getServerId(), currentTerm);

                    int voteGrantedNum = 0;
                    if (votedFor.intValue()==serverId) {
                        voteGrantedNum += 1;
                    }

                    for (Peer peer1 : peers) {
                        if (null!=peer1.getVoteGranted()&&peer1.getVoteGranted()) {
                            voteGrantedNum += 1;
                        }
                    }
                    log.info("server={},voteGrantedNum={}", serverId,voteGrantedNum);
                    if (voteGrantedNum > (peers.size()+1) / 2) {
                        becomeLeader();
                    }

                }else{
                    log.info("server={},Vote denied by server {} with term {}, my term is {}",serverId,
                            peer.getServer().getServerId(), voteResponse.getTerm(), currentTerm);

                    int negVoteGrantedNum = 0;
                    if (votedFor.intValue()!=serverId) {
                        negVoteGrantedNum += 1;
                    }

                    for (Peer peer1 : peers) {
                        if (null!=peer1.getVoteGranted()&&!peer1.getVoteGranted()) {
                            negVoteGrantedNum += 1;
                        }
                    }
                    log.info("server={},negVoteGrantedNum={}", serverId,negVoteGrantedNum);
                    if (negVoteGrantedNum > (peers.size()+1) / 2) {
                        stateMachine.tobeMaster(false);
                    }
                }
            }

        }

        @Override
        public void fail(Throwable throwable) {
            peer.setVoteGranted(new Boolean(false));
        }
    }


    private synchronized void becomeLeader() {
        if(state==NodeState.STATE_LEADER) {
            return;
        }

        //当一个领导人刚获得权力的时候，他初始化所有的 nextIndex 值为自己的最后一条日志的 index 加 1
        for (Peer peer : peers) {
            peer.setNextIndex(getLastLogIndex()+1);
        }
        log.info("Got majority vote, serverId={} become leader", serverId);
        state = NodeState.STATE_LEADER;

        stateMachine.tobeMaster(true);
        leaderId = serverId;
        // stop vote timer
        if (electionScheduledFuture != null && !electionScheduledFuture.isDone()) {
            electionScheduledFuture.cancel(true);
        }
        // start heartbeat timer
        startNewHeartbeat();
    }


    private void stepDown(long newTerm) {
        if (currentTerm > newTerm) {
            log.error("server={},currentTerm={},newTerm={},can't be happened",serverId,currentTerm,newTerm);
            return;
        }
        if (currentTerm < newTerm) {
            currentTerm = newTerm;
            leaderId = 0;
            votedFor = 0;
        }
        state = NodeState.STATE_FOLLOWER;
        stateMachine.tobeMaster(false);
        // stop heartbeat
        if (heartbeatScheduledFuture != null && !heartbeatScheduledFuture.isDone()) {
            heartbeatScheduledFuture.cancel(true);
        }
        resetElectionTimer();
    }
}
